package com.xl.flinkdemo.TableAPI;

import com.xl.flinkdemo.entity.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @ClassName Example
 * @Description TODO
 * @Author hxl
 * @Date 2021/5/21 11:37
 * Version 1.0
 **/
public class Example {
  public static void main(String[] args) throws Exception {

    StreamExecutionEnvironment environment =StreamExecutionEnvironment.getExecutionEnvironment();
    DataStreamSource<String> dataStreamSource =
        environment.readTextFile("D:\\studyspace\\flinkdemo\\src\\main\\resources\\sensors\\input.txt");

    //按照顺序,设置并行度为1
    environment.setParallelism(1);
    //将这个流转成对象实体
    DataStream<SensorReading> results = dataStreamSource.map(line->{
      String[] filter = line.split(",");
      SensorReading sensorReading = new SensorReading(filter[0],new Long(filter[1]),new Double(filter[2]));
      return sensorReading;
    });

    //开始进行tableAPI

    //1.得到table的env环境
    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(environment);

    //得到一张表
    Table table = tableEnv.fromDataStream(results);

    //调用TableAPI进行转换操作
    Table resultTable = table.select("id,sensors").where("id = 'sensor_1'");

    //执行SQL
    //注册这张表
    tableEnv.createTemporaryView("inputTable",results);
    String sql = "select id,sensors  from inputTable where id = 'sensor_1'";
    Table sqlResult = tableEnv.sqlQuery(sql);


    //打印SQL生成的table
    tableEnv.toAppendStream(sqlResult, Row.class).print("sql");

   tableEnv.toAppendStream(resultTable, Row.class).print("table");

    environment.execute();

  }
}
